package com.rec.smile.rpc.consumer;

import io.netty.bootstrap.Bootstrap;
import io.netty.channel.ChannelFuture;
import io.netty.channel.ChannelHandlerContext;
import io.netty.channel.ChannelInitializer;
import io.netty.channel.ChannelOption;
import io.netty.channel.EventLoopGroup;
import io.netty.channel.SimpleChannelInboundHandler;
import io.netty.channel.nio.NioEventLoopGroup;
import io.netty.channel.socket.SocketChannel;
import io.netty.channel.socket.nio.NioSocketChannel;

import com.rec.smile.rpc.api.RpcRequest;
import com.rec.smile.rpc.api.RpcResponse;
import com.rec.smile.rpc.api.codec.RpcDecoder;
import com.rec.smile.rpc.api.codec.RpcEncoder;

public class RpcClient extends SimpleChannelInboundHandler<RpcResponse> {

	
	private String serverAddress;
	
	private final Object obj = new Object();
	
	private RpcResponse rpcResponse;

	public RpcClient(String serverAddress) {
		super();
		this.serverAddress = serverAddress;
	}

    
	public RpcResponse send(RpcRequest rpcRequest){
		EventLoopGroup eventLoopGroup = new NioEventLoopGroup();
		try {
			Bootstrap bootstrap = new Bootstrap();
			bootstrap.group(eventLoopGroup).channel(NioSocketChannel.class)
			.handler(new ChannelInitializer<SocketChannel>() {

				@Override
				protected void initChannel(SocketChannel ch) throws Exception {
					 ch.pipeline().addLast(new RpcEncoder(RpcRequest.class), new RpcDecoder(RpcResponse.class), RpcClient.this);
				}
			}).option(ChannelOption.SO_KEEPALIVE, true);
			
			String[] array = serverAddress.split(":");
			ChannelFuture channelFuture = bootstrap.connect(array[0], Integer.parseInt(array[1])).sync();
			channelFuture.channel().writeAndFlush(rpcRequest).sync();
			
			synchronized (obj) {
				obj.wait();
			}
			
			if(rpcResponse != null){
				channelFuture.channel().close().sync();
			}
			
			return rpcResponse;
			
		} catch (Exception e) {
			e.printStackTrace();
		}finally{
			eventLoopGroup.shutdownGracefully();
		}
		return null;
	}


	@Override
	public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause)
			throws Exception {
		
		ctx.close();
	}


	@Override
	protected void channelRead0(ChannelHandlerContext ctx, RpcResponse msg)
			throws Exception {
		this.rpcResponse = msg;
		synchronized (obj) {
			obj.notifyAll();
		}
	}

}
